package com.chinaums.format;

import com.chinaums.format.option.SftpCsvOption;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.FactoryUtil;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

/**
 * @author 86185
 */
public class SftpCsvFormatFactory implements DeserializationFormatFactory {

    public static final ConfigOption<String> COLUMN_DELIMITER = ConfigOptions.key("column-delimiter")
            .stringType()
            .defaultValue("003")
            .withDescription("列分隔符 003 表示 \003 ");

    public static final ConfigOption<String> LINE_DELIMITER = ConfigOptions.key("line-delimiter")
            .stringType()
            .noDefaultValue()
            .withDescription("行分割符 ... ");

    public static final ConfigOption<Boolean> IGNORE_PARSE_ERRORS = ConfigOptions.key("ignore-parse-errors")
            .booleanType()
            .defaultValue(false)
            .withDescription("忽略错误行 ...");


    @Override
    public String factoryIdentifier() {
        return "sftp-csv";
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        return Collections.emptySet();
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(COLUMN_DELIMITER);
        options.add(IGNORE_PARSE_ERRORS);
        options.add(LINE_DELIMITER);
        return options;
    }

    @Override
    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) {


        //实现自定义验证逻辑
        FactoryUtil.validateFactoryOptions(this, formatOptions);
        // 创建并返回 DecodingFormat
        return new SftpCsvFormat(getFtpReadOptions(formatOptions));

    }


    private SftpCsvOption getFtpReadOptions(ReadableConfig formatOptions) {
        return SftpCsvOption.builder().setColumnDelimiter(formatOptions.get(COLUMN_DELIMITER)).setIgnoreParseErrors(formatOptions.get(IGNORE_PARSE_ERRORS)).setLineDelimiter(formatOptions.get(LINE_DELIMITER)).build();
    }


}
